package chenxu.test.jms;

import java.io.Closeable;
import java.io.File;
import java.io.IOException;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.blob.BlobTransferPolicy;
import org.apache.activemq.blob.BlobUploader;
import org.apache.activemq.command.ActiveMQBlobMessage;
import org.apache.activemq.command.MessageId;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import backtype.storm.contrib.jms.JmsProvider;


public class BlobMsgProducer implements Closeable {
	
	private  String ftpUser;
	private  String ftpPasswd;
	private  String ftpHost;
	private  JmsProvider jmsProvider;
	
	private static Log logger = LogFactory.getLog("");
	private MessageProducer producer;
	private Session session;
	private Connection connection;
			
	
	public BlobMsgProducer(String ftpUser, String ftpPasswd, String ftpHost,
			JmsProvider jmsProvider) {
		super();
		this.ftpUser = ftpUser;
		this.ftpPasswd = ftpPasswd;
		this.ftpHost = ftpHost;
		this.jmsProvider = jmsProvider;
		
		try {
			initJms();
		} catch (JMSException e) {
			logger.error("init jms error",e);
		} catch (Exception e) {
			logger.error("init jms error",e);
		}
	}
	
	private void initJms() throws JMSException, Exception {
		connection = jmsProvider.connectionFactory().createConnection();
		connection.start();
		
		session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		producer = session.createProducer(jmsProvider.destination());
		producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
	}

	public void sendMsg(String localFilePath, String remoteFileDir) {

			
		ActiveMQBlobMessage message = new ActiveMQBlobMessage();
		
		
		BlobTransferPolicy transferPolicy = new BlobTransferPolicy();
		BlobUploader uploader = new BlobUploader(transferPolicy, new File(localFilePath));

		String uploadUrl = " ftp://"+this.ftpUser+":"+this.ftpPasswd+"@"+this.ftpHost+remoteFileDir;
		transferPolicy.setUploadUrl(uploadUrl );

		message.setMessageId(new MessageId(localFilePath));
		message.setBlobUploader(uploader);

		try {
			producer.send(message);
			
		} catch (JMSException e) {
			logger.error("send msg error",e);
		} 
		
	}
	public ActiveMQBlobMessage sendMsg4Test(String localFilePath, String remoteFileDir) {
		
		
		ActiveMQBlobMessage message = new ActiveMQBlobMessage();
		
		
		BlobTransferPolicy transferPolicy = new BlobTransferPolicy();
		BlobUploader uploader = new BlobUploader(transferPolicy, new File(localFilePath));
		
		String uploadUrl = " ftp://"+this.ftpUser+":"+this.ftpPasswd+"@"+this.ftpHost+remoteFileDir;
		transferPolicy.setUploadUrl(uploadUrl );
		
		message.setMessageId(new MessageId(localFilePath));
		message.setBlobUploader(uploader);
		
		try {
			producer.send(message);
			return message;
			
		} catch (JMSException e) {
			logger.error("send msg error",e);
			return null;
		} 
		
	}
	

	@Override
	public void close() throws IOException {
		try {
			session.close();
			connection.close();
		} catch (JMSException e) {
			throw new IOException(e);
		}
	}

}
